package client

import (
	"bytes"
	"encoding/json"
	"github.com/sirupsen/logrus"
	http2 "net/http"
	"strings"
	"sync"
	"watcher/client/config"
	"watcher/client/watcher"
)

type Client struct {
	watchers   map[string]watcher.Watcher
	reportChan chan *watcher.WatchData
	failedChan chan *watcher.WatchData
	Done       chan struct{}
	config     config.Conf
}

// RegisterWatcher 注册需要监听的指标
func (c *Client) RegisterWatcher(watchers ...watcher.WatcherRegister) {
	logrus.Debugf("开始注册监听器...")
	var wt watcher.Watcher
	var err error
	var key string
	for _, w := range watchers {
		key, wt, err = w(c.config)
		if err != nil {
			logrus.Warnf("监听器[%s]注册失败！", key)
			break
		}
		c.watchers[key] = wt
		logrus.Debugf("监听器[%s]注册成功！", key)
	}
}

// Listen 启动所有watcher，监听系统数据，并将数据扇入reportChan中
func (c *Client) Listen() {
	// 创建channels 进行扇入
	wLen := len(c.watchers)
	channels := make([]<-chan *watcher.WatchData, wLen)
	for k, w := range c.watchers {
		ch := w.Watch()
		logrus.Debugf("开始监控[%s]指标！", k)
		channels = append(channels, ch)
	}
	c.fadeIn(channels...)
}

// Close 关闭client
func (c *Client) Close() {
	for _, w := range c.watchers {
		w.Close()
	}
}

// fadeIn 将多个通道的数据扇入client的reportChan中
func (c *Client) fadeIn(channels ...<-chan *watcher.WatchData) {
	var wg sync.WaitGroup
	wg.Add(len(channels))
	for _, ch := range channels {
		go func(ch <-chan *watcher.WatchData) {
			defer wg.Done()
			for v := range ch {
				if v.Tags == nil {
					v.Tags = map[string]string{
						"host": c.config.Host,
					}
				} else {
					v.Tags["host"] = c.config.Host
				}
				select {
				case <-c.Done:
					return
				case c.reportChan <- v: // 将多个通道的数据通过goroutine写入一个通道
				}
			}
		}(ch)
	}
	go func() {
		wg.Wait()
		close(c.reportChan) // 当所有的通道关闭后，关闭扇出通道
	}()
}

// Report 处理reportChan中的数据，将数据上报到server
func (c *Client) Report() {
	l := len(c.watchers)
	for i := 0; i < l; i++ {
		go c.report()
	}
}

func (c *Client) report() {
	for {
		select {
		case d := <-c.reportChan:
			b, err := json.Marshal(d)
			if err != nil {
				continue
			}
			url := strings.TrimRight(c.config.Server.Url, "/") + "/write"
			resp, err := http2.Post(url, "application/json", bytes.NewBuffer(b))
			if err != nil {
				logrus.Errorf("上报数据失败：%v", err)
			} else {
				logrus.Debugf("上报数据结果：[%d][%v]", resp.StatusCode, resp)
				_ = resp.Body.Close()
			}
		}
	}
}

func (c *Client) Log() {
	go func() {
		for {
			select {
			case d := <-c.reportChan:
				logrus.Debugf("监控[%s]指标数据：%s", d.Measurement, d)
			case <-c.Done:
				break
			}
		}
	}()
}

func NewClient(conf config.Conf) *Client {
	return &Client{
		watchers:   make(map[string]watcher.Watcher),
		reportChan: make(chan *watcher.WatchData, 10),
		failedChan: make(chan *watcher.WatchData, 100),
		config:     conf,
	}
}
